package chan.module.project.hw.mrs.kafka.v350;

import chan.module.project.hw.mrs.kafka.v350.security.LoginUtil;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.ListTopicsResult;
import org.apache.kafka.clients.admin.TopicListing;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Collection;
import java.util.Properties;

public class MRSKafkaV350TestMain {
    private static final Logger LOG = LoggerFactory.getLogger(MRSKafkaV350TestMain.class);
    private static final String USER_PRINCIPAL = "swap_mrs_uat@HADOOP.COM";
    private static final String KEYTAB_FILE_PATH = "/data/tongtech/TongDXP3181/Node-Agent/mrs-all-config/swap_mrs_uat.keytab";
    private static final String KER5CONFIG_FILE_PATH = "/data/tongtech/TongDXP3181/Node-Agent/mrs-all-config/KrbClient/config/krb5.conf";
    private static final String BOOTSTRAP_SERVERS = "10.242.60.229:21007";

    public static void main(String[] args){
        try {
            LoginUtil.securityPrepare(USER_PRINCIPAL,KEYTAB_FILE_PATH,KER5CONFIG_FILE_PATH);
            Properties props = new Properties();
            props.put("bootstrap.servers",BOOTSTRAP_SERVERS);
            try (AdminClient admin = AdminClient.create(props)) {
                ListTopicsResult topics = admin.listTopics();
                Collection<TopicListing> listings = topics.listings().get();

                LOG.info("Kafka Topics:");
                for (TopicListing listing : listings) {
                    LOG.info(listing.name());
                }
            }
        } catch (Exception e) {
            LOG.error("test kafka fail:",e);
        }

    }
}